package realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import realtime.bean.TradeOrderCountWithCategory;

import realtime.func.AsyncDIMFunction;
import realtime.util.ClickHouseUtil;
import realtime.util.DateFormatUtil;
import realtime.util.MyKafkaUtil;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.TimeUnit;

/**
 * @author MengX
 * @create 2023/3/25 15:43:13
 */
public class DwsTradeOrderCountWithCategoryWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //设置状态后端

//        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().setCheckpointTimeout(10000L);
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/dws/220926");

        //设置HDFS用户信息
        //System.setProperty("HADOOP_USER_NAME", "atguigu");

        //kafka读取数据
        String topic = "dwd_trade_order_detail";
        String groupId = "dwd_trade_order_detail_categroy_window";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));
        SingleOutputStreamOperator<JSONObject> jsonDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                if (value != null) {
                    out.collect(JSON.parseObject(value));
                }
            }
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
            @Override
            public long extractTimestamp(JSONObject element, long recordTimestamp) {
                return element.getLong("create_time");
            }
        }));
        SingleOutputStreamOperator<TradeOrderCountWithCategory> map = jsonDS.keyBy(line -> line.getString("user_id"))
                .map(new RichMapFunction<JSONObject, TradeOrderCountWithCategory>() {
                    ValueState<String> valueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("userct", String.class);
                        StateTtlConfig build = new StateTtlConfig.Builder(Time.seconds(5))
                                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                                .build();
                        valueStateDescriptor.enableTimeToLive(build);
                        valueState = getRuntimeContext().getState(valueStateDescriptor);
                    }

                    @Override
                    public TradeOrderCountWithCategory map(JSONObject value) throws Exception {
                        String lastct = valueState.value();
                        String curt = value.getString("create_time");
                        String orderId = value.getString("order_id");
                        String courseId = value.getString("course_id");
                        BigDecimal originAmount = value.getBigDecimal("origin_amount");
                        HashSet<String> orderIdSet = new HashSet<>();
                        orderIdSet.add(orderId);
                        long uc = 0L;
                        if (lastct == null) {
                            uc = 1L;
                            valueState.update(curt);
                        } else if (!lastct.equals(curt)) {
                            uc = 1L;
                            valueState.update(curt);
                        }
                        return TradeOrderCountWithCategory.builder()
                                .orderId(orderIdSet)
                                .courseId(courseId)
                                .orderAmount(originAmount)
                                .orderUc(uc)
                                .build();
                    }
                });
        SingleOutputStreamOperator<TradeOrderCountWithCategory> subjectId = AsyncDataStream.unorderedWait(map, new AsyncDIMFunction<TradeOrderCountWithCategory>("DIM_COURSE_INFO") {
            @Override
            public void join(TradeOrderCountWithCategory input, JSONObject dimInfo) {
                input.setSubjectId(dimInfo.getString("SUBJECT_ID"));
            }

            @Override
            public String getKey(TradeOrderCountWithCategory input) {
                return input.getCourseId();
            }
        }, 60, TimeUnit.SECONDS);

        SingleOutputStreamOperator<TradeOrderCountWithCategory> subjectDS = AsyncDataStream.unorderedWait(subjectId, new AsyncDIMFunction<TradeOrderCountWithCategory>("DIM_BASE_SUBJECT_INFO") {
            @Override
            public void join(TradeOrderCountWithCategory input, JSONObject dimInfo) {
                input.setId(dimInfo.getString("CATEGORY_ID"));
                input.setSubjectName(dimInfo.getString("SUBJECT_NAME"));

            }

            @Override
            public String getKey(TradeOrderCountWithCategory input) {
                return input.getSubjectId();
            }
        }, 60, TimeUnit.SECONDS);

        SingleOutputStreamOperator<TradeOrderCountWithCategory> categoryName = AsyncDataStream.unorderedWait(subjectDS, new AsyncDIMFunction<TradeOrderCountWithCategory>("DIM_BASE_CATEGORY_INFO") {
            @Override
            public void join(TradeOrderCountWithCategory input, JSONObject dimInfo) {
                input.setName(dimInfo.getString("CATEGORY_NAME"));
            }

            @Override
            public String getKey(TradeOrderCountWithCategory input) {
                return input.getId();
            }
        }, 60, TimeUnit.SECONDS);

        SingleOutputStreamOperator<TradeOrderCountWithCategory> resultDS = categoryName.keyBy(TradeOrderCountWithCategory::getId).window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10))).reduce(new ReduceFunction<TradeOrderCountWithCategory>() {
            @Override
            public TradeOrderCountWithCategory reduce(TradeOrderCountWithCategory value1, TradeOrderCountWithCategory value2) throws Exception {
                Set<String> orderId = value1.getOrderId();
                orderId.addAll(value2.getOrderId());
                value1.setOrderId(orderId);
                value1.setOrderUc(value1.getOrderUc() + value2.getOrderUc());
                value2.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                return value1;
            }
        }, new WindowFunction<TradeOrderCountWithCategory, TradeOrderCountWithCategory, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow window, Iterable<TradeOrderCountWithCategory> input, Collector<TradeOrderCountWithCategory> out) throws Exception {
                TradeOrderCountWithCategory next = input.iterator().next();
                int size = next.getOrderId().size();
                String end = DateFormatUtil.toYmdHms(window.getEnd());
                String start = DateFormatUtil.toYmdHms(window.getStart());
                long ts = System.currentTimeMillis();
                next.setOrderct((long) size);
                next.setStt(start);
                next.setEnt(end);
                next.setTs(ts);
                out.collect(next);
            }
        });

        resultDS.print("resultDS>>>>>>>>>>>>>>>>>>>>>>>");

        resultDS.addSink(ClickHouseUtil.getSinkFunction("insert into dws_trade_order_count_window values(?,?,?,?,?,?,?,?)"));

        env.execute("DwsTradeOrderCountWindow");
    }
}
